package com.shujia.sql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object Demo06EventTime {
  def main(args: Array[String]): Unit = {

    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() // 使用Blink的SQL解析器
      .inStreamingMode() // 开启流处理模式
      .build()
    // 构建Table Env
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    bsEnv.setParallelism(1)

    // 构建Source表 动态表
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE click_table (
          | `user` STRING,
          | cTime TIMESTAMP(3), -- 数据中自带的时间 可以作为 事件时间
          | url STRING,
          | WATERMARK FOR cTime AS cTime - INTERVAL '5' SECOND -- 设置5s的水位线
          |) WITH (
          | 'connector' = 'kafka',
          | 'topic' = 'click3',
          | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
          | 'properties.group.id' = 'test1',
          | 'format' = 'csv',
          | 'csv.field-delimiter' = ',',
          | 'csv.ignore-parse-errors' = 'true',
          | 'scan.startup.mode' = 'earliest-offset'
          |)
          |""".stripMargin)


    bsTableEnv
      .executeSql(
        """
          |create table click_cnt_mysql_sink(
          |   `user` String,
          |   startT TIMESTAMP(3),
          |   endT TIMESTAMP(3),
          |   cnt BIGINT
          |) WITH (
          |   'connector' = 'jdbc',
          |   'url' = 'jdbc:mysql://master:3306/student?useSSL=false',
          |   'table-name' = 'click_cnt',
          |   'username' = 'root',
          |   'password' = '123456'
          |)
          |""".stripMargin)


    // 每隔5s统计最近一个小时用户的点击次数

    bsTableEnv
      .executeSql(
        """
          |insert into click_cnt_mysql_sink
          |select user
          |       ,HOP_START(cTime, INTERVAL '5' SECONDS, INTERVAL '1' HOURS) as startT
          |       ,HOP_END(cTime, INTERVAL '5' SECONDS, INTERVAL '1' HOURS) as endT
          |       ,count(url) as cnt
          |from click_table
          |group by user,HOP(cTime, INTERVAL '5' SECONDS, INTERVAL '1' HOURS)
          |""".stripMargin)

  }

}
